package cn.linkai.hadoop.mydefine;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.Iterator;

public class TradeStat {

    public static void main(String[] args) {

        Configuration conf=new Configuration();
        Job job;
        Job sortJob;
        try {

            job=Job.getInstance(conf,"tradeStat");
            job.setJarByClass(TradeStat.class);
            sortJob=Job.getInstance(conf,"tradeSort");

            job.setMapperClass(TradeStatMapper.class);
            job.setReducerClass(TradeStatReducer.class);

            sortJob.setMapperClass(TradeSortMapper.class);
            sortJob.setReducerClass(TradeSortReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(TradeBean.class);

            job.setOutputKeyClass(TradeBean.class);
            job.setOutputKeyClass(NullWritable.class);

            sortJob.setOutputKeyClass(TradeBean.class);
            sortJob.setOutputValueClass(NullWritable.class);

            job.setInputFormatClass(TextInputFormat.class);

            FileInputFormat.addInputPath(job,new Path("E:\\hadoop\\hadoopdata\\tradestat"));
            FileOutputFormat.setOutputPath(job,new Path("E:\\hadoop\\hadoopdata\\outtradestat201804271156"));

            FileInputFormat.addInputPath(sortJob,new Path("E:\\hadoop\\hadoopdata\\outtradestat201804271156"));
            FileOutputFormat.setOutputPath(sortJob,new Path("E:\\hadoop\\hadoopdata\\outtradestat201804271156_2"));

            FileSystem system=FileSystem.get(conf);
            boolean isexit=system.exists(new Path("E:\\hadoop\\hadoopdata\\outtradestat201804271156"));
            if (isexit){
                system.delete(new Path("E:\\hadoop\\hadoopdata\\outtradestat201804271156"),true);
            }
            boolean state1=job.waitForCompletion(true);
            if(state1){
                System.out.println(sortJob.waitForCompletion(true)?"success":"fail");
            }else{
                System.out.println("fail");
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

    }

    public static class TradeStatMapper extends Mapper<LongWritable,Text,Text,TradeBean>{

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] infos=value.toString().split("\t");
            if(infos.length==4){
                Text userName=new Text(infos[0]);
                TradeBean tradeBean=new TradeBean();
                tradeBean.set(userName,new DoubleWritable(Double.valueOf(infos[1])),new DoubleWritable(Double.valueOf(infos[2])));
                context.write(userName,tradeBean);
                System.out.println(tradeBean);
            }else{
                return;
            }
        }
    }

    public static class TradeStatReducer extends Reducer<Text,TradeBean,TradeBean,NullWritable>{
        @Override
        protected void reduce(Text key, Iterable<TradeBean> values, Context context) throws IOException, InterruptedException {

            Iterator<TradeBean> iterator=values.iterator();
            double incomes=0;
            double outcomes=0;
            while (iterator.hasNext()){
                TradeBean bean=iterator.next();
                System.out.println("-----------"+bean);
                incomes+=bean.getIncome().get();
                outcomes+=bean.getOutcome().get();
            }
            TradeBean outBean=new TradeBean();
            outBean.set(key,new DoubleWritable(incomes),new DoubleWritable(outcomes));
            context.write(outBean,NullWritable.get());
        }
    }

    public static class TradeSortMapper extends Mapper<LongWritable,Text,TradeBean,NullWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] infos=value.toString().split("\t");
            TradeBean bean=new TradeBean();
            bean.set(new Text(infos[0]),new DoubleWritable(Double.valueOf(infos[1])),
                    new DoubleWritable(Double.valueOf(infos[2])));
            System.out.println(bean);
            context.write(bean,NullWritable.get());
        }
    }

    public static class TradeSortReducer extends Reducer<TradeBean,NullWritable,TradeBean,NullWritable>{
        @Override
        protected void reduce(TradeBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            System.out.println(key);

            context.write(key,values.iterator().next());
        }
    }
}
